package sameHostLinkGraph;

import java.io.IOException;
import java.util.Arrays;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import removeDanglingPage.ReadBytes;

import mapred.AbstractMapper;
import mapred.util.BytesBuilder;

public class SameHost_Mapper extends AbstractMapper<BytesWritable, BytesWritable, BytesWritable, BytesWritable> {

	@Override
	public void map(BytesWritable key, BytesWritable value,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		ReadBytes k = new ReadBytes(key);
		ReadBytes link = new ReadBytes(value);
		BytesBuilder gamma = new BytesBuilder();
		byte[] host = k.nextBytes(8);
		while(link.hasNextUrl()) {
			byte[] alpha = link.nextUrl();
			byte[] beta = Arrays.copyOf(alpha, 8);
			if(Arrays.equals(host, beta)) {
				gamma.append(alpha);
			}
		}
		output.collect(key, new BytesWritable(gamma.close()));
	}

}
